package com.sh.data.engine.application.service.datadev;

import cn.hutool.core.convert.Convert;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.sh.data.engine.application.model.dto.datadev.FlinkDatabaseResponse;
import com.sh.data.engine.application.model.request.datadev.*;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.ServletUtils;
import com.sh.data.engine.domain.base.model.PageResult;
import com.sh.data.engine.domain.base.model.UserContext;
import com.sh.data.engine.domain.datadev.flink.model.domain.*;
import com.sh.data.engine.domain.datadev.flink.model.param.SaveFlinkTaskNodeParam;
import com.sh.data.engine.domain.datadev.flink.service.FlinkJobService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkTaskV1Service;
import com.sh.data.engine.domain.shims.db.model.PreviewDataDomain;
import com.sh.data.engine.domain.util.ConvertUtil;
import jakarta.servlet.http.HttpServletResponse;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * @author: mengzheng.mhc
 * @date: 2024/8/9 10:50
 */
@Slf4j
@Service
public class FlinkTaskApplicationService {

    @Autowired
    private FlinkTaskV1Service flinkTaskV1Service;

    @Autowired
    private FlinkJobService flinkJobService;

    public boolean saveTaskNode(SavaFlinkTaskNodeRequest request) {
        SaveFlinkTaskNodeParam convert = Convert.convert(SaveFlinkTaskNodeParam.class, request);
        flinkTaskV1Service.saveFlinkTaskNode(convert);
        return true;
    }

    public boolean saveSqlTaskScript(SaveScriptRequest request) {
        String userId = UserContext.getUserId();
        Long projectId = ServletUtils.getProjectId();
        SaveScriptDomain saveScriptDomain = ConvertUtil.copyProperties(request, SaveScriptDomain.class);
        flinkTaskV1Service.saveSqlTaskScript(saveScriptDomain, userId, projectId);
        return true;
    }

    public boolean saveSource(SaveSourceRequest request) {
        List<SaveSourceRequest.Source> sources = request.getSources();
        long count = sources.stream().map(i -> i.getMappingTableName()).distinct().count();
        if (sources.size() != count) {
            throw new BusinessException("源表存在同名映射表");
        }
        sources.forEach(
            source -> {
                final List<FlinkKafkaSourceConfigDomain.PartitionOffset> partitionOffsetList =
                    source.getPartitionOffsetList();
                // SpecificOffset
                if (source.getKafkaSource().getOffset() == 4) {
                    if (partitionOffsetList == null || partitionOffsetList.size() == 0) {
                        throw new BusinessException("分区参数不能为空");
                    }
                }
            });

        String userId = UserContext.getUserId();
        Long projectId = ServletUtils.getProjectId();

        SaveSourceDomain convert = Convert.convert(SaveSourceDomain.class, request);
        flinkTaskV1Service.saveSource(convert, userId, projectId);
        return true;
    }

    public boolean saveSink(SaveSinkRequest request) {
        String userId = UserContext.getUserId();
        Long projectId = ServletUtils.getProjectId();
        SaveSinkDomain convert = Convert.convert(SaveSinkDomain.class, request);
        flinkTaskV1Service.saveSink(convert, userId, projectId);
        return true;
    }

    public boolean saveDim(SaveDimRequest request) {
        String userId = UserContext.getUserId();
        Long projectId = ServletUtils.getProjectId();
        SaveDimDomain convert = Convert.convert(SaveDimDomain.class, request);
        flinkTaskV1Service.saveDim(convert, userId, projectId);
        return true;
    }

    public boolean saveEnv(SaveEnvRequest request) {
        String userId = UserContext.getUserId();
        Long projectId = ServletUtils.getProjectId();
        SaveEnvDomain convert = Convert.convert(SaveEnvDomain.class, request);
        flinkTaskV1Service.saveEnv(convert, userId, projectId);
        return true;
    }

    public boolean saveSchedule(SaveScheduleRequest request) {
        String userId = UserContext.getUserId();
        Long projectId = ServletUtils.getProjectId();
        SaveScheduleDomain convert = Convert.convert(SaveScheduleDomain.class, request);
        flinkTaskV1Service.saveSchedule(convert, userId, projectId);
        return true;
    }

    public FlinkTaskDetailDomain getTaskDetail(Long nodeId, Long versionId) {
        return flinkTaskV1Service.getTaskDetail(nodeId, versionId);
    }

    public String getSqlTaskScript(Long nodeId, Long versionId) {
        return flinkTaskV1Service.getSqlTaskScrpit(nodeId, versionId);
    }

    public List<FlinkSourceDomain> getSourceDetail(Long nodeId, Long versionId) {
        return flinkTaskV1Service.getSourceDetail(nodeId, versionId);
    }

    public List<FlinkSinkDomain> getSinkDetail(Long nodeId, Long versionId) {
        return flinkTaskV1Service.getSinkDetail(nodeId, versionId);
    }

    public List<FlinkDimDomain> getDimDetail(Long nodeId, Long versionId) {
        return flinkTaskV1Service.getDimDetail(nodeId, versionId);
    }

    public String getEnvDetail(Long nodeId, Long versionId) {
        return flinkTaskV1Service.getEnvDetail(nodeId, versionId);
    }

    public FlinkScheduleDomain getScheduleDetail(Long nodeId, Long versionId) {
        return flinkTaskV1Service.getScheduleDetail(nodeId, versionId);
    }

    public PageResult<FlinkSqlTaskVersionDomain> getSqlTaskVersion(Long id, Integer pageNum, Integer pageSize) {
        IPage<FlinkSqlTaskVersionDomain> sqlTaskVersion =
            flinkTaskV1Service.getSqlTaskVersion(id, pageNum, pageSize);
        return PageResult.convert(sqlTaskVersion);
    }

    public List<String> getDsType(Integer businessType) {
        return flinkTaskV1Service.getDsType(businessType);
    }

    public List<FlinkDatabaseResponse> getDatabasesByDsType(String dsType) {
        Long projectId = ServletUtils.getProjectId();
        List<FlinkDatabaseDomain> databaseDomains = flinkTaskV1Service.getDatabasesByDsType(dsType, projectId);
        return ConvertUtil.copyProperties(databaseDomains, FlinkDatabaseResponse.class);
    }

    public Set<String> getTopicAndColumns(Long dsId) {
        return flinkTaskV1Service.getTopics(dsId);
    }

    public List<FlinkTableDomain> getTables(Long dsId, String schema, Integer isPartition) {
        return flinkTaskV1Service.getTables(dsId, ServletUtils.getProjectId(), schema, isPartition);
    }

    public List<FlinkTableFieldDomain> getTableFields(Long dsId, String tableName) {
        return flinkTaskV1Service.getTableFields(dsId, tableName);
    }

    public List<String> getHbaseColumnFamilyNames(Long dsId, String tableName) {
        return flinkTaskV1Service.getHbaseColumnFamilyNames(dsId, tableName);
    }

    public boolean rollbackVersion(Long nodeId, Long versionId) {
        return flinkTaskV1Service.rollbackVersion(nodeId, versionId);
    }

    public FlinkCheckGrammarDomain checkGrammar(String sql) {
        return flinkTaskV1Service.checkGrammar(sql);
    }

    public List<HiveTableInfoDomain> getPartitionTableList() {
        Long projectId = ServletUtils.getProjectId();
        return flinkTaskV1Service.getPartitionTableList(projectId);
    }

    public List<String> getPartitionInfoByTblName(String tableName) {
        Long projectId = ServletUtils.getProjectId();
        return flinkTaskV1Service.getPartitionInfoByTblName(tableName, projectId);
    }

    public List<FlinkSourceTableInfoDomain> getSqlTaskSourceTables(Long nodeId) {
        return flinkTaskV1Service.getSqlTaskSourceTables(nodeId);
    }

    public void downloadSqlTaskTemplate(Long nodeId, String tableName, HttpServletResponse response) {
        flinkTaskV1Service.downloadSqlTaskTemplate(nodeId, tableName, response);
    }

    public List<List<String>> uploadDebugData(MultipartFile file, Map<String, String[]> parameterMap) {
        if (!parameterMap.containsKey("nodeId") || parameterMap.get("nodeId").length == 0) {
            throw new BusinessException("nodeId参数必填");
        }

        if (!parameterMap.containsKey("tableName") || parameterMap.get("tableName").length == 0) {
            throw new BusinessException("tableName参数必填");
        }

        Long nodeId = Long.valueOf(parameterMap.get("nodeId")[0]);
        String tableName = parameterMap.get("tableName")[0];

        if (!StringUtils.endsWithIgnoreCase(file.getOriginalFilename(), ".csv")) {
            throw new BusinessException("仅支持csv格式的文件");
        }

        return flinkTaskV1Service.uploadDebugData(nodeId, tableName, file);
    }

    public List<List<String>> collectOnlineData(Long nodeId, String tableName, Integer maxCount) {
        return flinkTaskV1Service.collectOnlineData(nodeId, tableName, maxCount);
    }

    public FlinkDebugDomain startDebug(Long nodeId) {
        return flinkTaskV1Service.startDebug(nodeId);
    }

    public boolean stopDebug(Long jobId) {
        return flinkTaskV1Service.stopDebug(jobId);
    }

    public FlinkJobLogDomain getLog(Long jobId) {
        return flinkTaskV1Service.getLog(jobId);
    }

    public PreviewDataDomain previewData(Long dsId, String tableName) {
        return flinkTaskV1Service.previewData(dsId, tableName);
    }

    public List<String> previewKafkaData(Long dsId, String topic) {
        return flinkTaskV1Service.previewKafkaData(dsId, topic);
    }
}